package com.ylh.untils;

import com.ylh.bean.PtVisitCnt;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.io.network.ConnectionManager;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.conf.ConfigurationManager;
import org.apache.hadoop.hbase.util.Bytes;


public class SinkHbase extends RichSinkFunction<PtVisitCnt> {
    Connection connection=null;
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        org.apache.hadoop.conf.Configuration config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "hadoop-single:2181");  // Until 2.x.y versions
        connection = ConnectionFactory.createConnection(config);
    }

    @Override
    public void close() throws Exception {
        super.close();
        connection.close();
    }

    @Override
    public void invoke(PtVisitCnt value, Context context) throws Exception {
        super.invoke(value, context);
        //获取表名
        Table table = connection.getTable(TableName.valueOf("pt_list"));
        System.out.println("invoke>>");
        Put put = new Put(Bytes.toBytes(value.getPage_id()+"-"+value.getPt()));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("cnt"), Bytes.toBytes(value.getCnt()));
        table.put(put);
    }
}
